package spring.cloud.tasks.executor.executor.task;

import lombok.extern.slf4j.Slf4j;
import spring.cloud.tasks.common.Threads;
import spring.cloud.tasks.executor.executor.Executor;
import spring.cloud.tasks.executor.executor.task.execute_runtime_context.TaskItemExecuteRuntimeContext;
import spring.cloud.tasks.executor.executor.task.execute_runtime_context.TasksExecuteRuntimeContext;
import spring.cloud.tasks.executor.executor.task.execute_statistics.ExecuteStatistics;
import spring.cloud.tasks.executor.executor.task.task_item_execute.Status;
import spring.cloud.tasks.executor.executor.task.task_item_execute.TaskItemExecuteRenewalInfoService;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;

/**
 * 任务执行状态维护线程
 * 1collect task execute status and execute result thread
 * 2超时或者异常的任务分片终止
 * 3活跃的任务分片续期
 */
@Slf4j
public class TaskStateMaintenanceThread extends Thread {
    private final TaskExecutor taskExecutor;

    public TaskStateMaintenanceThread(TaskExecutor taskExecutor) {
        this.taskExecutor = taskExecutor;
    }

    @Override
    public void run() {
        Lock lock = taskExecutor.lock;
        try {
            if (Executor.getNeedStopFlag()) {
                return;
            }
            if (taskExecutor.isNeedStop()) {
                return;
            }
            //这里得不到TaskItemExecute 对象。故不在这里进行判断
            if (taskExecutor.lock.tryLock(10, TimeUnit.SECONDS)) {
                try {
                    statusCheckAndRenewal();
                } catch (Exception ex) {
                    log.error(ex.getMessage(), ex);
                } finally {
                    //当获取锁成功时最后一定要记住finally去关闭锁
                    lock.unlock();   //释放锁
                }
            }
        } catch (InterruptedException ex) {
            log.error(ex.getMessage(), ex);
        }
    }

    /**
     * 如果当前任务片不在进行分配 那么则不在续期。出现这种情况 那么其他的执行器可以覆盖。
     *
     * @throws Exception
     */
    private void statusCheckAndRenewal() throws Exception {
        /**
         * 如果是状态维护则不能新建默认对象。必须使用原生的map对象判断
         */
        TasksExecuteRuntimeContext.ergodicTaskExecuteRuntimeContextAndHandleBusiness(taskExecutor.getTaskId(), new TasksExecuteRuntimeContext.Business() {
            @Override
            public void doBusiness(ConcurrentHashMap<String, TaskItemExecuteRuntimeContext> taskItemIndexToTaskItemExecuteRuntimeContextMap, TaskItemExecuteRuntimeContext taskItemExecuteRuntimeContext) {
                try {
                    statusCheckAndRenewal(taskItemIndexToTaskItemExecuteRuntimeContextMap, taskItemExecuteRuntimeContext);
                    if (taskItemExecuteRuntimeContext != null && taskItemExecuteRuntimeContext.isFinished() && !taskItemExecuteRuntimeContext.isReported) {
                        taskItemExecuteRuntimeContext.isReported = true;
                        //可能异步进行汇报状态
                        reportTaskItemExecuteRuntimeContext(taskItemExecuteRuntimeContext);
                    }
                } catch (Throwable e) {
                    log.error(e.getMessage(), e);
                }
            }
        });
    }

    /**
     * 推出两种续期方式。1业务方法，同步的进行续期。2业务方法每执行一个代码点，则更新自己的心跳时间，证明自己还活着，由维护程序进行统一的续期。
     * 1需要自己进行主动续期，性能相比2低一点。
     *
     * @param taskItemExecuteRuntimeContext
     */
    private void statusCheckAndRenewal(ConcurrentHashMap<String, TaskItemExecuteRuntimeContext> taskItemIndexToTaskItemExecuteRuntimeContextMap, TaskItemExecuteRuntimeContext taskItemExecuteRuntimeContext) {
        //优先保证上下文信息被清理掉
        if (System.currentTimeMillis() >= taskItemExecuteRuntimeContext.allowDeleteTaskItemExecuteRuntimeContextTimeStamp) {
            taskItemIndexToTaskItemExecuteRuntimeContextMap.remove(taskItemExecuteRuntimeContext.taskStage$type$index);
            //
            if (taskItemExecuteRuntimeContext.needStop) {
                return;
            } else {
                taskItemExecuteRuntimeContext.needStop = true;
                taskItemExecuteRuntimeContext.needStopMessage = "已经超过锁结束时间了";
                return;
            }
        } else {
            //还没有注册续期信息，那么则没有分配对应的线程进行处理
            if (!taskItemExecuteRuntimeContext.isRegistered()) {
                //如果没有注册，根据allowDeleteTaskItemExecuteRuntimeContextTimeStamp进行上下文信息过期清理
                return;
            } else {
                //如果任务完成了肯定就不用续期
                if (taskItemExecuteRuntimeContext.isFinished()) {
                    return;
                } else {
                    //已经满足了分片停止的条件,则什么都不用做了
                    if (taskItemExecuteRuntimeContext.needStop) {
                        return;
                    } else {
                        statusCheckAndRenewal(taskItemExecuteRuntimeContext);
                    }

                }
            }

        }
    }

    /**
     * 状态检测,如果状态正常续期。（有线程进行处理了+状态不是完成状态）
     */
    private void statusCheckAndRenewal(TaskItemExecuteRuntimeContext taskItemExecuteRuntimeContext) {
        //分片停止指令优先
        boolean isAssigned = taskExecutor.getExecutor().getTaskMap().isAssigned(taskItemExecuteRuntimeContext);
        if (!isAssigned) {
            taskItemExecuteRuntimeContext.needStop = true;
            taskItemExecuteRuntimeContext.needStopMessage = "任务分配给其他执行器了或者任务已经取消";
            return;
        }
        //进行续期信息校验
        long nowDateTime = System.currentTimeMillis();
        if (nowDateTime > taskItemExecuteRuntimeContext.getMaxAllowRunningTime()) {
            taskItemExecuteRuntimeContext.needStop = true;
            taskItemExecuteRuntimeContext.needStopMessage = "已经超过最大的任务允许运行时间了";
            return;
        }
        Long maxRenewalTime = taskItemExecuteRuntimeContext.getMaxRenewalTime();
        if (nowDateTime > maxRenewalTime) {
            taskItemExecuteRuntimeContext.needStop = true;
            taskItemExecuteRuntimeContext.needStopMessage = "已经超过最大的续期时间了";
            return;
        }

        if (nowDateTime > taskItemExecuteRuntimeContext.getTimeoutDateTime().getTime()) {
            taskItemExecuteRuntimeContext.needStop = true;
            taskItemExecuteRuntimeContext.needStopMessage = "已经超过超时时间了";
            return;
        }
        if (nowDateTime > taskItemExecuteRuntimeContext.getLockDateTime().getTime()) {
            taskItemExecuteRuntimeContext.needStop = true;
            taskItemExecuteRuntimeContext.needStopMessage = "已经超过锁结束时间了";
            return;
        }
        // 1线程信息
        String threadName = taskItemExecuteRuntimeContext.getThreadName();
        boolean createdThread = taskItemExecuteRuntimeContext.isCreatedThread();
        if (!createdThread) {

        } else {
            Thread thread = Threads.getThreadByhreadName(threadName);
            if (thread == null) {
                //可能线程没有创建或者已经终止了
                if (nowDateTime - taskItemExecuteRuntimeContext.getBeginTime() >= 1000 * 60) {
                    taskItemExecuteRuntimeContext.setStatus(Status.EXCEPTION_END);
                    taskItemExecuteRuntimeContext.setMessage("线程已经不存在了");
                    return;
                } else {
                    return;//等待下次，在这不能进行设置为任务结束
                }
            }
            if (!thread.isAlive()) {
                Threads.sleep();//status fix 可能任务执行的代码真正处于修改状态中
                if (taskItemExecuteRuntimeContext.getStatus() == Status.RUNNING) {
                    taskItemExecuteRuntimeContext.setStatus(Status.EXCEPTION_END);
                    taskItemExecuteRuntimeContext.setMessage("线程处于非活跃状态 但是还存在");
                    return;
                } else {
                    return;
                }

            }
        }
        // 4任务执行信息的汇报检查
        long lastReportDate = taskItemExecuteRuntimeContext.lastReportDate.getTime();
        if (lastReportDate == taskItemExecuteRuntimeContext.getBeginTime()) {
            //才开始
            return;
        }
        if (nowDateTime - lastReportDate <= 1000 * 60 * 1) {//续期时间间隔点存在问题
            //主动通知状态管理任务还存存活
            TaskItemExecuteRenewalInfoService taskItemExecuteInfoService = taskExecutor.getTaskItemExecuteInfoService();
            TaskItemExecuteRuntimeContext.RenewalTimeInfo renewalTimeInfo = taskItemExecuteRuntimeContext.calculateNewRenewalInfo();
            Boolean renewal = taskItemExecuteInfoService.renewal(taskItemExecuteRuntimeContext, renewalTimeInfo);
            if (renewal == null) {
                return;
            } else {
                if (renewal) {
                    taskItemExecuteRuntimeContext.setMaxAllowRunningTime(renewalTimeInfo.maxAllowRunningTime);
                    taskItemExecuteRuntimeContext.setMaxRenewalTime(renewalTimeInfo.maxRenewalTime);
                    return;
                } else {
                    taskItemExecuteRuntimeContext.needStop = true;
                    taskItemExecuteRuntimeContext.needStopMessage = "续期失败";
                    return;
                }
            }
        }
        /**
         * <pre>
         *     MYSQL 自增值重新生成方法
         * CREATE TABLE `xxxx` (
         * `id` bigint(20) NOT NULL AUTO_INCREMENT COMMENT '自增主键',
         * `create_time` datetime(3) NOT NULL DEFAULT current_timestamp(3) COMMENT '创建时间',
         * `update_time` datetime(3) NOT NULL DEFAULT current_timestamp(3) ON UPDATE current_timestamp(3) COMMENT '更新时间',
         * PRIMARY KEY (`id`) USING BTREE,
         * KEY `idx_create_time` (`create_time`) USING BTREE,
         * KEY `idx_update_time` (`update_time`) USING BTREE
         * ) ENGINE=InnoDB AUTO_INCREMENT=5 DEFAULT CHARSET=utf8mb4 COMMENT='表注释';
         *
         * 方法一(效率高，谨慎使用)：
         * 清空表数据并重置id：
         * truncate table table_name;
         * delete from table_name;
         * alter table table_name auto_increment= 1;
         *
         * ORACLE 序列生成方法。
         * create sequence seq_xxxx
         * increment by 1
         * start with 1
         * minvalue 1
         * maxvalue 99999999(最大值建议小于1亿，如果超过1亿那么在内存计算分片的时候特别多。一般分片范围为1k,1w.如果单条数据关联处理的数据很大。建议设置的更小，100.)
         * order
         * cache 10000
         * cycle;
         * 这里需要取数据的min-max,根据[min-max]/范围单位，按照绝整10,100,整1000以及其他值进行范围拆分。
         * 当前数据拆分可能支持1维拆分，或者二维组合拆分（比如按照部门或者产品分类，在这个维度下再进行数据范围拆分）。
         * 上诉拆分完毕后，再分多线程无区分的小批次处理。（当然支持多阶段）
         * </pre>
         */
        //捞取数据最长的时间（DB数据捞取数据，同时写入到临时表）
        //以及如何构造序列升序序列（这个时间最长是两个小时）
        if (nowDateTime - lastReportDate >= 1000 * 60 * 32) {//如果一个任务32分钟还没有进行汇报状态,可以立即进行停止了。
            taskItemExecuteRuntimeContext.needStop = true;
            taskItemExecuteRuntimeContext.needStopMessage = "即使任务线程你还活着也通知你不要执行了";
            return;
        }
        //2-10分钟内的不主动进行续期 因为状态检查是1分钟进行一次检查。要是任务活跃那么更新时间间隔差必然小于2分钟内
        return;
    }

    protected void reportTaskItemExecuteRuntimeContext(TaskItemExecuteRuntimeContext taskItemExecuteRuntimeContext) {
        int successCount = 0;
        int errorCount = 0;
        int status = taskItemExecuteRuntimeContext.getStatus();
        if (status == Status.SUCCESS) {
            successCount = 1;
        } else {
            errorCount = 1;
        }
        // 执行次数加1
        String taskId = taskItemExecuteRuntimeContext.getTaskId();
        String executorId = taskItemExecuteRuntimeContext.getExecutorId();

        // 只要有出错和失败的分片，就认为是处理失败; 否则认为处理成功
        if (errorCount == 0) {
            ExecuteStatistics.incrementProcessSuccessCount(taskId, successCount);
        } else {

            ExecuteStatistics.incrementProcessFailureCount(taskId, errorCount);
        }
    }
}
